/**
 * 使用worker_threads和SharedArrayBuffer实现的滑动窗口
 * 支持多线程共享内存
 */
const { isMainThread, Worker, workerData, parentPort } = require('worker_threads');

class SharedSlidingWindow {
  /**
   * 构造函数
   * @param {number} duration - 持续时间(秒)
   * @param {number} threshold - 阈值，超过该值判定为热点
   */
  constructor(duration, threshold) {
    // 超过10分钟的按10分钟
    if (duration > 600) {
      duration = 600;
    }
    
    // 为短时间窗口和长时间窗口设置不同的时间片配置
    if (duration <= 5) {
      this.windowSize = 5;
      this.timeMillisPerSlice = duration * 200;
    } else {
      this.windowSize = 10;
      this.timeMillisPerSlice = duration * 100;
    }
    
    this.threshold = threshold;
    // 保证存储在至少两个window
    this.timeSliceSize = this.windowSize * 2;
    
    // 初始化共享内存 - 为计数和时间戳使用足够大的缓冲区
    // timeSliceSize 个计数 + 2 个时间戳（秒部分） + 2 个时间戳（毫秒部分）
    this.sharedArrayBuffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT * (this.timeSliceSize + 4)); 
    this.sharedArray = new Int32Array(this.sharedArrayBuffer);
    
    // 为了更清晰的代码，存储时间戳的索引
    this.BEGIN_TIME_SEC_INDEX = this.timeSliceSize;
    this.BEGIN_TIME_MS_INDEX = this.timeSliceSize + 1;
    this.LAST_ADD_TIME_SEC_INDEX = this.timeSliceSize + 2;
    this.LAST_ADD_TIME_MS_INDEX = this.timeSliceSize + 3;
    
    // 初始化时间片数组
    this.reset();
  }
  
  /**
   * 初始化/重置窗口
   */
  reset() {
    const now = Date.now();
    
    // 将时间戳拆分为秒和毫秒部分，以适应Int32
    const secPart = Math.floor(now / 1000);
    const msPart = now % 1000;
    
    // 使用原子操作写入时间戳
    Atomics.store(this.sharedArray, this.BEGIN_TIME_SEC_INDEX, secPart);
    Atomics.store(this.sharedArray, this.BEGIN_TIME_MS_INDEX, msPart);
    Atomics.store(this.sharedArray, this.LAST_ADD_TIME_SEC_INDEX, secPart);
    Atomics.store(this.sharedArray, this.LAST_ADD_TIME_MS_INDEX, msPart);
    
    // 初始化时间片数组
    for (let i = 0; i < this.timeSliceSize; i++) {
      Atomics.store(this.sharedArray, i, 0);
    }
  }
  
  /**
   * 计算当前所在的时间片的位置
   * @returns {number} - 当前时间片索引
   */
  locationIndex() {
    const now = Date.now();
    
    // 重建完整的时间戳
    const beginSecPart = Atomics.load(this.sharedArray, this.BEGIN_TIME_SEC_INDEX);
    const beginMsPart = Atomics.load(this.sharedArray, this.BEGIN_TIME_MS_INDEX);
    const beginTimestamp = (beginSecPart * 1000) + beginMsPart;
    
    const lastSecPart = Atomics.load(this.sharedArray, this.LAST_ADD_TIME_SEC_INDEX);
    const lastMsPart = Atomics.load(this.sharedArray, this.LAST_ADD_TIME_MS_INDEX);
    const lastAddTimestamp = (lastSecPart * 1000) + lastMsPart;
    
    // 如果当前的key已经超出一整个时间片了，那么就直接初始化
    if (now - lastAddTimestamp > this.timeMillisPerSlice * this.windowSize) {
      this.reset();
      return 0;
    }
    
    let index = Math.floor(((now - beginTimestamp) / this.timeMillisPerSlice) % this.timeSliceSize);
    if (index < 0) {
      return 0;
    }
    return index;
  }
  
  /**
   * 清除从index开始的旧时间片数据
   * @param {number} index - 当前索引
   */
  clearFromIndex(index) {
    for (let i = 1; i <= this.windowSize; i++) {
      let j = index + i;
      if (j >= this.windowSize * 2) {
        j -= this.windowSize * 2;
      }
      Atomics.store(this.sharedArray, j, 0);
    }
  }
  
  /**
   * 增加count个数量，并判断是否超过阈值
   * @param {number} count - 增加的数量
   * @returns {boolean} - 是否超过阈值
   */
  addCount(count) {
    // 当前自己所在的位置，是哪个小时间窗
    const index = this.locationIndex();
    
    // 清空旧的时间片数据
    this.clearFromIndex(index);
    
    // 在当前时间片里增加计数 (使用原子操作)
    const newCount = Atomics.add(this.sharedArray, index, count);
    
    // 计算窗口内的总和
    let sum = 0;
    // 加上当前时间片
    sum += newCount + count; // 因为Atomics.add返回的是旧值，所以需要加上count
    
    // 加上前面几个时间片
    for (let i = 1; i < this.windowSize; i++) {
      const prevIndex = (index - i + this.timeSliceSize) % this.timeSliceSize;
      sum += Atomics.load(this.sharedArray, prevIndex);
    }
    
    // 更新最后添加时间，拆分为秒和毫秒部分
    const now = Date.now();
    Atomics.store(this.sharedArray, this.LAST_ADD_TIME_SEC_INDEX, Math.floor(now / 1000));
    Atomics.store(this.sharedArray, this.LAST_ADD_TIME_MS_INDEX, now % 1000);
    
    // 判断是否超过阈值
    return sum >= this.threshold;
  }
  
  /**
   * 获取整个滑动窗口的共享内存缓冲区
   * 可以传递给其他线程
   */
  getSharedArrayBuffer() {
    return this.sharedArrayBuffer;
  }
  
  /**
   * 从现有的共享内存缓冲区恢复滑动窗口对象
   * @param {SharedArrayBuffer} buffer - 共享内存缓冲区
   * @param {number} duration - 持续时间(秒)
   * @param {number} threshold - 阈值
   */
  static fromSharedArrayBuffer(buffer, duration, threshold) {
    const window = new SharedSlidingWindow(duration, threshold);
    // 替换内部缓冲区但保留索引常量
    const BEGIN_TIME_SEC_INDEX = window.BEGIN_TIME_SEC_INDEX;
    const BEGIN_TIME_MS_INDEX = window.BEGIN_TIME_MS_INDEX;
    const LAST_ADD_TIME_SEC_INDEX = window.LAST_ADD_TIME_SEC_INDEX;
    const LAST_ADD_TIME_MS_INDEX = window.LAST_ADD_TIME_MS_INDEX;
    
    window.sharedArrayBuffer = buffer;
    window.sharedArray = new Int32Array(buffer);
    
    // 恢复索引常量
    window.BEGIN_TIME_SEC_INDEX = BEGIN_TIME_SEC_INDEX;
    window.BEGIN_TIME_MS_INDEX = BEGIN_TIME_MS_INDEX;
    window.LAST_ADD_TIME_SEC_INDEX = LAST_ADD_TIME_SEC_INDEX;
    window.LAST_ADD_TIME_MS_INDEX = LAST_ADD_TIME_MS_INDEX;
    
    return window;
  }
}

// 工作线程逻辑
if (!isMainThread) {
  try {
    // 确保workerData包含所有必需的字段
    if (!workerData || !workerData.buffer) {
      console.error('Worker missing required data');
      parentPort.postMessage({ error: 'Worker missing required data' });
      return;
    }
    
    const { buffer, duration, threshold, index } = workerData;
    
    // 安全检查
    if (!(buffer instanceof SharedArrayBuffer)) {
      console.error('Invalid SharedArrayBuffer');
      parentPort.postMessage({ error: 'Invalid SharedArrayBuffer' });
      return;
    }
    
    console.log(`工作线程 ${index} 初始化成功`);
    
    // 维护一个键到滑动窗口的映射
    const keyWindows = new Map();
    
    // 接收指令和数据
    parentPort.on('message', (message) => {
      try {
        // 处理关闭命令
        if (message.cmd === 'shutdown') {
          console.log(`工作线程 ${index} 收到关闭命令，正在退出...`);
          // 清空资源
          keyWindows.clear();
          // 退出工作线程
          process.exit(0);
          return;
        }
        
        // 处理添加计数命令
        if (message.cmd === 'addCount') {
          const { key, count, buffer: keyBuffer, interval, threshold } = message;
          
          if (!key) {
            console.error('收到的消息缺少键');
            return;
          }
          
          let window;
          
          // 如果提供了新的缓冲区，使用它
          if (keyBuffer instanceof SharedArrayBuffer) {
            window = SharedSlidingWindow.fromSharedArrayBuffer(keyBuffer, interval || 10, threshold || 100);
            keyWindows.set(key, window);
          } else {
            // 否则使用现有的
            window = keyWindows.get(key);
            // 如果没有现有的，使用工作线程初始化时的缓冲区（不理想但防止崩溃）
            if (!window) {
              window = SharedSlidingWindow.fromSharedArrayBuffer(buffer, duration || 10, threshold || 100);
              keyWindows.set(key, window);
            }
          }
          
          const result = window.addCount(count || 1);
          parentPort.postMessage({ key, isHot: result, count: count || 1 });
        }
      } catch (err) {
        console.error('处理消息时出错:', err);
        parentPort.postMessage({ error: `处理消息时出错: ${err.message}`, key: message.key });
      }
    });
    
    // 报告工作线程就绪
    parentPort.postMessage({ status: 'ready', index });
  } catch (err) {
    console.error('工作线程初始化错误:', err);
    if (parentPort) {
      parentPort.postMessage({ error: `工作线程初始化错误: ${err.message}` });
    }
  }
}

module.exports = SharedSlidingWindow;
